package com.neusoft.tf;

import com.neusoft.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Random;

public class TransformationApiMain {
    private static final Random random = new Random();


    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        定义数据源
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
                l -> {
                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
                    return line;
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.STRING);
        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource")
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] words = value.split(",");
                        WaterSensor sensor = new WaterSensor(words[0], words[1], words[2]);
                        if (sensor.getVc() % 2 == 1)
                            out.collect(sensor);
                        else {
                            out.collect(sensor);
                            out.collect(sensor);
                        }
                    }
                }, Types.POJO(WaterSensor.class))

                .print();
        env.execute();
    }


    public static void main2(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        定义数据源
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
                l -> {
                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
                    return line;
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.STRING);
        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource")
                .filter(line -> line.startsWith("1"))
                .print();
        env.execute();
    }


    /**
     * TODO :Map映射算子
     *
     * @param args
     * @throws Exception
     */
    public static void main1(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
/**
 * TODO: 直接使用lambda 表达式，生成数据
 */
//        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
//                l -> {
//                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
//                    System.out.println(line);
//                    return line;
//                },
//                Long.MAX_VALUE,
//                RateLimiterStrategy.perSecond(1),
//                Types.STRING);
/**
 * TODO: 或者 使用定义静态类，生成数据
 */

        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource(
                l -> {
                    String line = ("" + (random.nextInt(3) + 1) + "," + System.currentTimeMillis() + "," + random.nextInt(90, 100));
                    System.out.println(line);
                    return line;
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.STRING);

        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "dataGeneratorSource")
                .map(new MyMapFunction())
                .print();
        env.execute();

    }

    public static class MyMapFunction implements MapFunction<String, WaterSensor> {
        @Override
        public WaterSensor map(String value) throws Exception {
            String[] words = value.split(",");
            WaterSensor sensor = new WaterSensor(words[0], words[1], words[2]);
            return sensor;
        }
    }
}
